package com.ganqiang1983.dbdiff;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

/**
 * 资源调度器
 */
public final class ScheduleManager
{
  private static Logger logger = Logger.getLogger(ScheduleManager.class);

  private static ScheduleManager sche = new ScheduleManager();

  public static ScheduleManager newInstance()
  {
    return sche;
  }

  // 根据类型判断该job属于哪个步骤
  public enum JOB_TYPE
  {
    DUMP, SORT, DIFF
  }

  // 本软件分为两种工作模式：[start模式|proceed模式]
  // 是否执行上次未完成的工作，如果执行(proceed)，则继续寻找未完成的工作
  // 如果执行(start)，则重新覆盖掉之前已经比较完成工作
  public boolean IS_PROCEED_MODE = true;
  // 是否删除中间过程遗留下来的文件，如果不删除，就需要关注检查文件合理性
  public boolean IS_DELETE_TEMPFILE = true;
  // 最终是否生成统计文件
  public boolean IS_CREATE_SUMMARYFILE = true;
  // 如果走分页查询的话，默认每页数量
  public int PAGE_SIZE = 1000;
  // dump消费者数量
  private int D_WORKER_COUNT;
  // sort消费者数量
  private int S_WORKER_COUNT;
  // diff消费者数量
  private int C_WORKER_COUNT;
  // dumper工作计划缓冲队列
  private final BlockingQueue<Job> D_QUEUE = new LinkedBlockingQueue<Job>();
  // sorter工作计划缓冲队列
  private final BlockingQueue<Job> S_QUEUE = new LinkedBlockingQueue<Job>();
  // comparer工作计划缓冲队列
  private final BlockingQueue<Job> C_QUEUE = new LinkedBlockingQueue<Job>();
  // 生产者
  private Scheduler scheduler;
  // 当前完工的数量数量
  private AtomicInteger CURRENT_COUNT = new AtomicInteger(0);
  // 给完工数量加把锁
  private Lock CURRENT_COUNT_LOCK = new ReentrantLock(false);

  /*
   * 将构造方法访问修饰符设为私有，禁止任意实例化。
   */
  private ScheduleManager()
  {
  }

  /**
   * setp 0 : set thread count
   */
  private void setDefaultWorkerCount()
  {
    int djobsize = scheduler.getDjobsize();
    int sjobsize = scheduler.getSjobsize();
    int cjobsize = scheduler.getCjobsize();
    if (djobsize == 1) {
      D_WORKER_COUNT = 1;
    } else if (D_WORKER_COUNT <= 0) { //如果启动时没有手工设置，默认为CPU核数
      D_WORKER_COUNT = Runtime.getRuntime().availableProcessors();
    }
    if (sjobsize == 1) {
      S_WORKER_COUNT = 1;
    } else if (S_WORKER_COUNT <= 0) { //如果启动时没有手工设置，默认为CPU核数
      S_WORKER_COUNT = Runtime.getRuntime().availableProcessors();
    }
    if (cjobsize == 1) {
      C_WORKER_COUNT = 1;
    } else if (C_WORKER_COUNT <= 0) { //如果启动时没有手工设置，默认为CPU核数
      C_WORKER_COUNT = Runtime.getRuntime().availableProcessors();
    }
    logger.info("[Main-Thread] Startup parameters : dumpThreadCount=[" + D_WORKER_COUNT + "], sortThreadCount="
        + "[" + S_WORKER_COUNT + "], compareThreadCount=[" + C_WORKER_COUNT + "]");
  }

  /**
   * step 1 : schedule
   */
  public void doSche()
  {
    long start = System.currentTimeMillis();
    scheduler = new Scheduler(JOB_TYPE.DUMP, D_QUEUE, S_QUEUE,
        C_QUEUE);
    scheduler.run();
    setDefaultWorkerCount();
    long end = System.currentTimeMillis();
    logger.debug("[Main-Thread] It takes [" + (end - start) + "]ms to execute [doSche] function.");
  }

  /**
   * step 2 : dump
   */
  public void doDump() throws InterruptedException
  {
    int djobsize = scheduler.getDjobsize();
    ExecutorService dumpWorker = Executors.newFixedThreadPool(D_WORKER_COUNT);
    for (int i = 0; i < djobsize; i++) {
      Worker worker = new Worker(JOB_TYPE.DUMP, D_QUEUE, S_QUEUE);
      dumpWorker.execute(worker);
    }
    dumpWorker.shutdown();
  }

  /**
   * step 3 : sort
   */
  public void doSort() throws InterruptedException
  {
    int sjobsize = scheduler.getSjobsize();
    ExecutorService sortWorker = Executors.newFixedThreadPool(S_WORKER_COUNT);
    for (int i = 0; i < sjobsize; i++) {
      Worker worker = new Worker(JOB_TYPE.SORT, S_QUEUE, C_QUEUE);
      sortWorker.execute(worker);
    }
    sortWorker.shutdown();
  }

  /**
   * step 4 : diff
   */
  public void doDiff() throws InterruptedException
  {
    int cjobsize = scheduler.getCjobsize();
    ExecutorService diffWorker = Executors.newFixedThreadPool(C_WORKER_COUNT);
    for (int i = 0; i < cjobsize; i++) {
      Worker worker = new Worker(JOB_TYPE.DIFF, C_QUEUE, CURRENT_COUNT, CURRENT_COUNT_LOCK);
      diffWorker.execute(worker);
    }
    diffWorker.shutdown();
  }

  /**
   * step 5 : statistic
   * @throws InterruptedException
   */
  public void doStat() throws InterruptedException
  {
    long start = System.currentTimeMillis();
    int cjobsize = scheduler.getCjobsize();
    while (true) {
      int currentcount = CURRENT_COUNT.get();
      if (cjobsize > 0 && cjobsize == currentcount) {
        boolean flag = FileMaker.writeSummaryFile();
        if (!flag) {
          System.exit(1);
        }
        break;
      }
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        logger.error("[Main-Thread] Waiting for statistical data failure : " + e.getMessage(), e);
        System.exit(1);
      }
    }
    long end = System.currentTimeMillis();
    logger.debug("[Main-Thread] It takes [" + (end - start) + "]ms to execute [doStat] function.");
  }

  /**
   * set commond options
   */
  private void setOptions(String param)
  {
    if (param.contains("D")) {
      D_WORKER_COUNT = Integer.valueOf(param.substring(1).toString());
    } else if (param.contains("S")) {
      S_WORKER_COUNT = Integer.valueOf(param.substring(1).toString());
    } else if (param.contains("C")) {
      C_WORKER_COUNT = Integer.valueOf(param.substring(1).toString());
    }
  }


  /**
   * commond
   */
  public static void main(String... args) throws Exception
  {
    long start = System.currentTimeMillis();
    ScheduleManager sm = ScheduleManager.newInstance();
    if (args.length == 0) {
      logger.error("[Main-Thread] Invalid parameter. You need to enter the one parameter at least.");
      System.exit(1);
    } else {
      if ("start".equalsIgnoreCase(args[0])) {
        sm.IS_PROCEED_MODE = false;
        logger.info("[Main-Thread] Starting DBDiff [start] mode.");
      } else {
        logger.info("[Main-Thread] Starting DBDiff [proceed] mode.");
      }
      if (args.length == 2) {
        sm.setOptions(args[1]);
      } else if (args.length == 3) {
        sm.setOptions(args[1]);
        sm.setOptions(args[2]);
      } else if (args.length >= 4) {
        sm.setOptions(args[1]);
        sm.setOptions(args[2]);
        sm.setOptions(args[3]);
      }
    }
    try {
      sm.doSche();
      sm.doDump();
      sm.doSort();
      sm.doDiff();
      sm.doStat();
      logger.info("[Main-Thread] DBDiff execution completed.");
    } catch (Exception e) {
      logger.error("[Main-Thread] DBDiff is interrupted : " + e.getMessage(), e);
      System.exit(1);
    }
    long end = System.currentTimeMillis();
    logger.debug("[Main-Thread] DBDiff spent a total of [" + (end - start) + "]ms.");
  }

}
